package com.fanli.bigdata.mytest

import java.sql.DriverManager

import com.fanli.bigdata.db.StateRedisConnections
import org.joda.time.format.DateTimeFormat
import org.json4s.jackson.JsonMethods._

import scala.collection.mutable.ArrayBuffer
import java.text.SimpleDateFormat
import java.util.{Date, Locale}

import scala.util.matching.Regex

/**
 * Created by liangdong.wu on 2016/12/8.
 */
object VarnishFun {
  def logStat(res: Array[(Realtime201611PUvKey,Realtime201611PUvValue)]):Unit= {
    val redisClient = new StateRedisConnections("115.159.45.213",6379)
    val redis = redisClient.redis
    val REDIS_EXPIR: Int = 24 * 60 * 60 * 2 //2day

    var url = "jdbc:mysql://115.159.45.213:3306/test?useUnicode=true&characterEncoding=utf8"
    val conn= DriverManager.getConnection(url,"hive","hive")
    var stmt = conn.createStatement()

    res.foreach(r=>{
      val prefix_key = "d11_" + r._1.day + "_"
      val prefix_detail_key = prefix_key + r._1.hour + "_"
      val prefix_detail_key_pv=prefix_key+"pv"

      val total_uvs = r._2.us.toSeq
      //日 uv
      redis.pfadd(prefix_key, total_uvs(0), total_uvs: _*)
      redis.expire(prefix_key,REDIS_EXPIR)

      //小时 uv
      redis.pfadd(prefix_detail_key, total_uvs(0), total_uvs: _*)
      redis.expire(prefix_detail_key,REDIS_EXPIR)

      //日累计 pv
      redis.incrby(prefix_detail_key_pv, r._2.count.toInt)
      redis.expire(prefix_detail_key_pv, REDIS_EXPIR)

      val  duvCount = redis.pfcount(prefix_key)
      val  huvCount = redis.pfcount(prefix_detail_key)
      val  sumPvCount = redis.get(prefix_detail_key_pv).getOrElse(0)
      println(duvCount + "\t" + huvCount +"\t"+ sumPvCount)

      val dSql = "INSERT INTO tb_varnish_realtime_flow(record_date, pv, uv, create_time) VALUES(\"" +
        r._1.day + "\"," + sumPvCount + "," + duvCount + ", NOW() )"+
        "ON DUPLICATE KEY UPDATE pv=" + sumPvCount + ",uv=" + duvCount
      //      println(dSql)
      stmt.addBatch(dSql)
    })
    stmt.executeBatch()
    stmt.close()
    conn.close
  }

  def MatchLogBatchFunc(msgs: Iterator[String]): Iterator[(Realtime201611PUvKey,Realtime201611PUvValue)] = {
    val arr = ArrayBuffer[(Realtime201611PUvKey,Realtime201611PUvValue)]()
    msgs.foreach(msg=>{
      val varnish = getLogFunc(msg)
      var uvSet = Set[String]()
      if (varnish.uid != "" || varnish.uid != "-"){
        uvSet += varnish.uid
      }
      arr += ((Realtime201611PUvKey(varnish.day,varnish.hour),Realtime201611PUvValue(uvSet,1L)))
    })
    arr.toIterator
  }

  def MatchLogSteamingFunc(msgs: Iterator[(String, String)]): Iterator[(Realtime201611PUvKey,Realtime201611PUvValue)] = {
    val arr = ArrayBuffer[(Realtime201611PUvKey,Realtime201611PUvValue)]()
    msgs.foreach(msg=>{
      val varnish = getLogFunc(msg._2)
      println(msg._1 + "\n" + msg._2)
      var uvSet = Set[String]()
      if (varnish.uid != "" || varnish.uid != "-"){
        uvSet += varnish.uid
      }
      arr += ((Realtime201611PUvKey(varnish.day,varnish.hour),Realtime201611PUvValue(uvSet,1L)))
    })
    arr.toIterator
  }

  def getLogFunc(log:String): Realtime201611 = {
//    val log = """111.19.32.224 - 104444908 [02/Nov/2016:17:43:54 +0800] GET http://fun.fanli.com/ HTTP/1.1 200 80 "http://m.fanli.com/taobao" "Fanli/5.4.2.11 (HUAWEI PLK-TL01H; Android 6.0; zh_CN; ID:2-104444908-67732835577999-12-0)" 0.011075258 miss pass 2168972125.2841786673.892491704 2168972125.2841786673.2733802942 21195b -"""
    val varnish_pattern = """(?:-|[^-]*) (?:-|[^ ]*) ([^ ]*) \[([^\]]*)\] [^ ]* ([^ ]*) [^ ]* [^ ]* (?:-|\d*) \"([^ ]*)\" \"([\s\S]*?)\" [\s\S]*""".r
    val agent_pattern="""(?:[\s\S]*?)ID:[0-9]-([0-9]{0,})-[0-9]{0,}-[\s\S]*""".r
    val sdf = new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss Z", Locale.US)
    val hourSdf = new SimpleDateFormat("yyyy-MM-dd HH:mm", Locale.US)
    var logTime: Date = new Date()
    var day = ""
    var hour = ""
    var uid = ""
    log match {
      case varnish_pattern(user_id, timeStr,path,refer, agent) => {
        try {
          logTime = sdf.parse(timeStr)
          val hourTime = hourSdf.format(logTime).split(" ")
          day = hourTime(0)
          hour =hourTime(1).split(":")(0)
          if( user_id=="-" && (agent != "-" || agent != "")){
            agent match {
              case agent_pattern(agent_id) => {
                uid = agent_id
              }
              case _ =>
            }
          }else{
            uid=user_id
          }
        } catch {
          case t: Throwable => println("error time,log===" + log)
        }
      }
      case _ =>
    }
    Realtime201611(day,hour,uid)
  }

  def varnishTestFun():Unit={
    val arr = ArrayBuffer[(Realtime201611PUvKey,Realtime201611PUvValue)]()
    val varnish_pattern = """(?:-|[^-]*) (?:-|[^ ]*) ([^ ]*) \[([^\]]*)\] [^ ]* ([^ ]*) [^ ]* [^ ]* (?:-|\d*) \"([^ ]*)\" \"([\s\S]*?)\" [\s\S]*""".r
    val agent_pattern="""(?:[\s\S]*?)ID:[0-9]-([0-9]{0,})-[0-9]{0,}-[\s\S]*""".r
    val path_referer_pattern = """.*(?:&|\?)u_?id=(\d+).*""".r
    val sdf = new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss Z", Locale.US)
    val hourSdf = new SimpleDateFormat("yyyy-MM-dd HH", Locale.US)
    val log = """111.19.32.224 - 104444908 [02/Nov/2016:17:43:54 +0800] GET http://fun.fanli.com/ HTTP/1.1 200 80 "http://m.fanli.com/taobao" "Fanli/5.4.2.11 (HUAWEI PLK-TL01H; Android 6.0; zh_CN; ID:2-104444908-67732835577999-12-0)" 0.011075258 miss pass 2168972125.2841786673.892491704 2168972125.2841786673.2733802942 21195b -"""
    val varnish_pattern(user_id, timeStr,path,refer, agent) = log
    println(user_id +"\t"+ timeStr +"\t" + agent)
    val agent_pattern(agent_id) = agent
    println(agent_id)
    var logTime: Date = new Date()
    var dt = ""
    var time = ""
    var userId = ""
    logTime = sdf.parse(timeStr)
    val hourTime = hourSdf.format(logTime).split(" ")
    dt = hourTime(0)
    time = hourTime(1)
    if( user_id=="-" && (agent != "-" || agent != "")){userId = agent_id}else{userId=user_id}
    var uvSet = Set[String]()
    if (userId != "" || userId != "-"){uvSet += userId}
    println(uvSet)
  }
}